#include "paxos.h"
#include "timeutils.h"
#include "broker_protocol.pb.h"
#include "paxos_protocol.pb.h"
#include "trace.h"
#include "method_thread.h"

#include <algorithm>

using namespace std;

// ensure total order
bool operator> (const Propnum &a, const Propnum &b)
{
    return a.n() > b.n() || (a.n() == b.n() && a.m() > b.m());
}

bool operator>= (const Propnum &a, const Propnum &b)
{
    return a.n() > b.n() || (a.n() == b.n() && a.m() >= b.m());
}

paxos::paxos(const string &primary, messaging *m, paxos_callback *cb)
    : primary_(primary), hermes_(m), instance_(0), behind_(false), cb_(cb)
{
    my_prop_.set_m(m->id());
    my_prop_.set_n(0); // TODO load logged proposals

    hermes_->subscribe("PAXOS.PREPAREREQ", this);
    hermes_->subscribe("PAXOS.PREPARERES", this);
    hermes_->subscribe("PAXOS.ACCEPTREQ", this);
    hermes_->subscribe("PAXOS.ACCEPTRES", this);
    hermes_->subscribe("PAXOS.DECIDEREQ", this);
    hermes_->subscribe("PAXOS.DECIDERES", this);

    // acceptor event handlers
    handlers_["PREPAREREQ"] = &paxos::handle_preparereq;
    handlers_["PREPARERES"] = &paxos::handle_prepareres;
    handlers_["ACCEPTREQ"] = &paxos::handle_acceptreq;
    handlers_["ACCEPTRES"] = &paxos::handle_acceptres;
    handlers_["DECIDEREQ"] = &paxos::handle_decidereq;
    handlers_["DECIDERES"] = &paxos::handle_decideres;

    pthread_mutex_init(&m_, NULL);
    pthread_cond_init(&accept_cv_, NULL);
    pthread_cond_init(&prepare_cv_, NULL);
    pthread_cond_init(&decide_cv_, NULL);
    pthread_cond_init(&behind_cv_, NULL);

    catchup_th_ = method_thread(this, false, &paxos::catchup);
    assert(catchup_th_ != 0);
}

paxos::~paxos()
{
    pthread_mutex_destroy(&m_);
    pthread_cond_destroy(&accept_cv_);
    pthread_cond_destroy(&prepare_cv_);
    pthread_cond_destroy(&decide_cv_);
    pthread_cond_destroy(&behind_cv_);
}

/**
 * Start a new instance of Paxos to agree on a proposed value. Note that, in
 * general Paxos, the proposed value is not neccessarily the chosen one.
 * However, in our scenario, as the primary is the only proposer, this can be
 * guaranteed. This function blocks until the consensus is made.
 */
string paxos::start(const vector<string> &mems, const string &val)
{
    assert(pthread_mutex_lock(&m_) == 0);
    unsigned instance = instance_ + 1;
    string va;
    members_ = mems;
    TDEBUG("starting paxos #%u", instance);
    while (true) {
        // XXX do we have to increment instance_ as well?
        TDEBUG("incrementing my_prop_.n (current: %d)", my_prop_.n());
        my_prop_.set_n(my_prop_.n() + 1); // increment my proposal number
        va.clear();
        vector<string> accepts;
        if (prepare(instance, mems, accepts, va)) {
            // we've received a majority of prepare_ok, now we send accept to
            // those we replied with ok
            if (va.size() == 0) {
                // no value's been accepted, so choose my own value
                va = val;
            }
            //decide_size_ = accept_size_ = accepts.size();
            if (accept(instance, accepts, va)) {
                // alright, the value has been decided
                decide(instance, accepts, va);
                break;
            } else {
                TWARN("no majority accept responses");
            }
        } else if (values_.size() >= instance && values_[instance - 1] != "") {
            // a special case of prepare()'s returning false
            TDEBUG("paxos #%u decided already. skipping accept", instance);
            break;
        } else {
            TWARN("no majority prepare responses");
        }
    }
    // instance_++;
    behind_ = false;
    assert(pthread_mutex_unlock(&m_) == 0);

    return va;
}

// assumes that the m_ mutex is held
bool paxos::prepare(unsigned instance, const vector<string> &mems,
        vector<string> &accepts, string &va)
{
    // send prepare messages to all those in ``mems''
    DataMessage req;
    PrepareRequest preq;

    preq.set_instance(instance);
    Propnum *pn = preq.mutable_pnum();
    *pn = my_prop_;
    req.set_subject("PAXOS.PREPAREREQ");
    req.set_body(preq.SerializeAsString());

    pbox_.watch(instance, mems.size());

    for (unsigned i = 0; i < mems.size(); ++i) {
        req.clear_dest();
        req.add_dest(mems[i]);
        TDEBUG("sending prepare to %s", mems[i].c_str());
        hermes_->deliver(&req);
    }

    pthread_mutex_unlock(&m_);
    int r = pbox_.wait(10000);
    pthread_mutex_lock(&m_);
    if (r == obj_bucket<pair<string, PrepareResponse *> >::FORCED_QUIT) {
        // we don't need to continue this paxos instance, as it has already
        // been decided by others some time earlier
        return false;
    } else if (r == obj_bucket<pair<string, PrepareResponse *> >::TIMEDOUT) {
        TWARN("not all recipients answered my prepare req");
    }
    TDEBUG("examining %u received prepare responses", pbox_.current_size());

    Propnum pa; // highest proposal ever accepted by others
    pa.Clear();
    va.clear();
    vector<pair<string, PrepareResponse *> > &replies = pbox_.items();
    // examine pbox_ to find those who accepted my prepare
    for (unsigned i = 0; i < replies.size(); ++i) {
        PrepareResponse *pr = replies[i].second;
        if (pr->accept()) {
            accepts.push_back(replies[i].first);
        }
        if (pr->has_accepted_val()) {
            if (va == "" || pr->accepted_pnum() > pa) {
                pa = pr->accepted_pnum();
                va = pr->accepted_val();
            }
        }
        delete pr;
    }
    pbox_.clear();
    TDEBUG("prepare_ok size: %u, mem sz: %u, pa (%s, %u), va: %s",
            accepts.size(), mems.size(), pa.m().c_str(), pa.n(), va.c_str());
    return (accepts.size() >= ((mems.size() >> 1) + 1));
}

// assumes that the m_ mutex is held
bool paxos::accept(unsigned instance, const vector<string> &mems,
        const string &val)
{
    DataMessage req;
    AcceptRequest areq;

    areq.set_instance(instance);
    areq.set_value(val);
    Propnum *pn = areq.mutable_pnum();
    *pn = my_prop_;
    req.set_subject("PAXOS.ACCEPTREQ");
    req.set_body(areq.SerializeAsString());

    abox_.watch(instance, mems.size());
    TDEBUG("accept #%u value: %s", instance, val.c_str());
    for (unsigned i = 0; i < mems.size(); ++i) {
        req.clear_dest();
        req.add_dest(mems[i]);
        TDEBUG("sending accept to %s", mems[i].c_str());
        hermes_->deliver(&req);
    }

    pthread_mutex_unlock(&m_);
    int r = abox_.wait(10000);
    pthread_mutex_lock(&m_);
    if (r == obj_bucket<AcceptResponse *>::TIMEDOUT) {
        TWARN("not all recipients answered my accept req");
    }

    TDEBUG("examining %u received accept responses", abox_.current_size());
    unsigned i = 0, accepts = 0;
    vector<AcceptResponse *> &replies = abox_.items();
    for (; i < replies.size(); ++i) {
        if (replies[i]->accept()) {
            accepts++;
        }
        delete replies[i];
    }
    abox_.clear();

    TDEBUG("accepts: %d, mem sz: %u", accepts, mems.size());
    return (accepts >= ((mems.size() >> 1) + 1));
}

void paxos::decide(unsigned instance, const vector<string> &mems,
        const string &v)
{
    DataMessage req;
    DecideRequest dreq;
    dreq.set_instance(instance);
    dreq.set_value(v);
    req.set_subject("PAXOS.DECIDEREQ");
    req.set_body(dreq.SerializeAsString());
    TDEBUG("value #%u decided: %s", instance, v.c_str());
    // save the current instance number because we might receive an incoming
    // decide request that updates my instance_ number

    dbox_.watch(instance, mems.size());
    for (unsigned i =  0; i < mems.size(); ++i) {
        req.clear_dest();
        req.add_dest(mems[i]);
        TDEBUG("sending decide to %s", mems[i].c_str());
        hermes_->deliver(&req);
    }

    // nasty hack to ensure all peers receive the decide
    // gonna enhance this in the future
    pthread_mutex_unlock(&m_);
    int r = dbox_.wait(10000);
    pthread_mutex_lock(&m_);
    if (r == obj_bucket<int>::TIMEDOUT) {
        TWARN("not all recipients answered my decide req");
    }
    dbox_.clear();
}

/**
 * Commit the value chosen by Paxos. Called upon receiving a decide message.
 * assuming m_ is held
 */
void paxos::commit(unsigned instance, const string &value)
{
    if (instance > instance_) {
        TDEBUG("setting highest instance to %u (original %u)", instance,
                instance_);
        instance_ = instance;
        if (instance > values_.size()) {
            TDEBUG("resize decided value vector from %u to %u", values_.size(),
                    instance);
            values_.resize(instance, "");
        }
        TDEBUG("paxos #%u decided on value %s", instance, value.c_str());
        values_[instance - 1] = value;
        pa_me_.Clear();
        pa_me_.set_m(hermes_->id());
        pp_seen_.Clear();
        pp_seen_.set_m(hermes_->id());
        va_me_.clear();

        if (cb_) {
            cb_->paxos_commit(instance, value);
        }
    }
}

void paxos::catchup()
{
    assert(pthread_mutex_lock(&m_) == 0);
    while (true) {
        while (!behind_) {
            TDEBUG("i am not behind; sleeping");
            pthread_cond_wait(&behind_cv_, &m_); 
        }
        TDEBUG("catching up. current paxos instance: %u", instance_);
        vector<string> primary_only;
        primary_only.push_back(primary_);
        assert(pthread_mutex_unlock(&m_) == 0); // cause start() needs the lock
        while (start(primary_only, ".") == ".")
            ;
        assert(pthread_mutex_lock(&m_) == 0);
        break;
    }
    assert(pthread_mutex_unlock(&m_) == 0);
}

void paxos::handle_preparereq(const DataMessage *data)
{
    assert(pthread_mutex_lock(&m_) == 0);
    PrepareRequest req;
    if (req.ParseFromString(data->body())) {
        TDEBUG("prepare req from %s inst %u, me %u", data->src().c_str(),
                req.instance(), instance_);
        DataMessage resp;
        PrepareResponse presp;
        resp.add_dest(data->src());
        resp.set_subject("PAXOS.PREPARERES");
        presp.set_instance(req.instance());
        if (req.instance() <= instance_) {
            // old instance
            const string &decided = values_[req.instance() - 1];
            presp.set_accept(false);
            presp.set_old(true);
            TINFO("tell %s the chosen value for old paxos instance #%u: %s",
                    data->src().c_str(), req.instance(), decided.c_str());
            presp.set_accepted_val(decided);
        } else if (req.instance() == instance_ + 1) {
            presp.set_old(false);
            if (req.pnum() > pp_seen_) {
                // i promise not to accept proposals that are less than req.num
                // and send back an OK
                pp_seen_ = req.pnum();
                presp.set_instance(instance_ + 1);
                presp.set_accept(true);
                TDEBUG("promising not to accept proposals < %u", pp_seen_.n());
                if (va_me_ != "") {
                    presp.set_accepted_val(va_me_);
                    Propnum *pnum = presp.mutable_accepted_pnum();
                    *pnum = pa_me_;
                }
            } else {
                // reject this proposal, or just silently ignore it??
                // for now we reject it
                presp.set_accept(false);
                TDEBUG("rejecting proposal numbered %u, high: %u",
                        req.pnum().n(), pp_seen_.n());
            }
        } else {
            // a prepare from the future.
            // this means other paxos nodes have made one or more decisions
            // which i haven't learned about yet. to solve this problem, i'll
            // first reject this prepare (because i am not ready), and then
            // start my own paxos instance to catch up. hopefully, i will learn
            // the decision in the prepare phase because the acceptor will
            // notice i sent an old instance and will reply with the chosen
            // value for that instance.
            presp.set_old(false);
            presp.set_accept(false);
            TWARN("i am behind; signaling catchup thread");
            behind_ = true;
            pthread_cond_signal(&behind_cv_);
        }
        resp.set_body(presp.SerializeAsString());
        hermes_->deliver(&resp);
    }
    assert (pthread_mutex_unlock(&m_) == 0);
}

void paxos::handle_prepareres(const DataMessage *data)
{
    assert(pthread_mutex_lock(&m_) == 0);
    PrepareResponse *resp = new PrepareResponse;
    if (resp->ParseFromString(data->body())) {
        TDEBUG("prepare response instance %u", resp->instance());
        if (resp->old()) {
            TDEBUG("i am behind. value for paxos #%u: %s", resp->instance(),
                    resp->accepted_val().c_str());
            commit(resp->instance(), resp->accepted_val());
            TDEBUG("aborting ongoing prepare");
            // pthread_cond_signal(&prepare_cv_);
            pbox_.stop_waiting();
            delete resp;
        //} else if (resp->instance() == instance_ + 1) {
        } else {
            if (pbox_.add(resp->instance(), pair<string, PrepareResponse *>(
                        data->src(), resp))) {
                TDEBUG("prepare resp #%u is exactly what we want",
                        resp->instance());
                if (resp->accept()) {
                    if (resp->accept()) {
                        TDEBUG("prepare ok #%u from %s", resp->instance(),
                                data->src().c_str());
                    } else {
                        TDEBUG("prepare rej #%u from %s", resp->instance(),
                                data->src().c_str());
                    }
                }
            } else {
                TWARN("prepare resp #%u comes too late", resp->instance());
                delete resp;
            }
            // it is possible that a prepare response may come after the
            // prepare phase ends. we don't handle this as a special case here
            // because signaling the conditional variable will have no effect.
        }
    } else {
        delete resp;
    }
    assert (pthread_mutex_unlock(&m_) == 0);
}

void paxos::handle_acceptreq(const DataMessage *data)
{
    assert(pthread_mutex_lock(&m_) == 0);
    AcceptRequest req;
    DataMessage resp;
    AcceptResponse aresp;
    resp.add_dest(data->src());
    resp.set_subject("PAXOS.ACCEPTRES");
    if (req.ParseFromString(data->body())) {
        TDEBUG("accept req #%u from %s %s(%u)", req.instance(),
                data->src().c_str(), req.pnum().m().c_str(), req.pnum().n());
        aresp.set_instance(req.instance());
        if (req.instance() == instance_ + 1) {
            TDEBUG("highest prepare seen: %s(%u)", pp_seen_.m().c_str(),
                    pp_seen_.n());
            if (req.pnum() >= pp_seen_) {
                // i can accept this proposal
                va_me_ = req.value();
                pa_me_ = req.pnum();
                aresp.set_accept(true);
            } else {
                aresp.set_accept(false);
            }
            aresp.set_old(false);
        } else {
            TWARN("unrecognized paxos instance #%u me %u", req.instance(),
                    instance_);
            aresp.set_accept(false);
            aresp.set_old(true);
        }
        resp.set_body(aresp.SerializeAsString());
        hermes_->deliver(&resp);
    }
    assert(pthread_mutex_unlock(&m_) == 0);
}

void paxos::handle_acceptres(const DataMessage *data)
{
    assert(pthread_mutex_lock(&m_) == 0);
    AcceptResponse *resp = new AcceptResponse;
    if (resp->ParseFromString(data->body())) {
        // remove duplicates in pbox and abox
        TDEBUG("accept response from %s #%u me %u", data->src().c_str(),
                resp->instance(), instance_);
        //if (resp->instance() == instance_ + 1) {
        if (abox_.add(resp->instance(), resp)) {
            if (resp->accept()) {
                TDEBUG("accept ok #%u from %s", resp->instance(),
                        data->src().c_str());
            } else {
                TDEBUG("accept rej #%u from %s", resp->instance(),
                        data->src().c_str());
            }
        } else {
            TWARN("unrecognized accept response #%u (me: %u)", resp->instance(),
                    instance_);
            delete resp;
        }
    } else {
        delete resp;
    }
    assert(pthread_mutex_unlock(&m_) == 0);
}

void paxos::handle_decidereq(const DataMessage *data)
{
    assert(pthread_mutex_lock(&m_) == 0);
    DecideRequest req;
    if (req.ParseFromString(data->body())) {
        TDEBUG("decide req from %s #%u", data->src().c_str(), req.instance());
        commit(req.instance(), req.value());
        DataMessage reply;
        DecideResponse dres;
        reply.set_subject("PAXOS.DECIDERES");
        reply.add_dest(data->src());
        dres.set_instance(req.instance());
        dres.set_status("OK");
        reply.set_body(dres.SerializeAsString());
        TDEBUG("sending decide ack #%u to %s", req.instance(),
                data->src().c_str());
        hermes_->deliver(&reply);
    }
    assert(pthread_mutex_unlock(&m_) == 0);
}

void paxos::handle_decideres(const DataMessage *data)
{
    assert(pthread_mutex_lock(&m_) == 0);
    DecideRequest resp;
    if (resp.ParseFromString(data->body())) {
        TDEBUG("decide ack #%u from %s", resp.instance(), data->src().c_str());
        if (dbox_.add(resp.instance(), data->src())) {
            TDEBUG("expected decide ack #%u", resp.instance());
        } else {
            TWARN("unrecognized decide ack #%u", resp.instance());
        }
    }
    assert(pthread_mutex_unlock(&m_) == 0);
}

void paxos::on_message(const DataMessage *data)
{
    const string &subject = data->subject();
    if (subject.compare(0, 6, "PAXOS.") == 0) {
        string subsub = subject.substr(6);
        if (handlers_.find(subsub) != handlers_.end()) {
            ev_handler hn = handlers_[subsub];
            (this->*hn)(data);
        } else {
            printf("unknown paxos message: %s\n", subsub.c_str());
        }
    } else {
        printf("irrelevant message\n");
    }
}

